/*
 * Copyright (c) www.bugull.com
 */

package com.bugull.farm.core.server;

import com.bugull.farm.core.cache.PropertiesCache;
import com.bugull.farm.core.utils.StringUtil;
import com.bugull.farm.core.utils.ThreadUtil;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.logging.log4j.*;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

/**
 *
 * @author Frank Wen(xbwen@hotmail.com)
 */
public class ZKClient {

    private final static Logger logger = LogManager.getLogger(ZKClient.class);

    private CuratorFramework curator;
    private String ip;
    private AtomicLong channelCount;
    private ExecutorService executor;
    private String workerType;

    private ZKClient(){
        channelCount = new AtomicLong();
        executor = Executors.newSingleThreadExecutor();
    }

    private static class Holder {
        final static ZKClient instance = new ZKClient();
    }

    public static ZKClient getInstance(){
        return Holder.instance;
    }

    public CuratorFramework getCurator() {
        return curator;
    }

    public void setIp(String workerType, String ip) {
        this.ip = ip;
        this.workerType = workerType;
        //check connection
        if(curator == null){
            logger.error("You should connect to ZK Server first!");
            return;
        }

     // create parent node if not exists
     		Stat stat = null;
     		try {
     			stat = curator.checkExists().forPath(workerType);
     		} catch (Exception ex) {
     			logger.error(ex.getMessage(), ex);
     		}
     		if (stat == null) {
     			try {
     				curator.create().withMode(CreateMode.PERSISTENT)
     						.forPath(workerType, null);
     			} catch (Exception ex) {
     				logger.error(ex.getMessage(), ex);
     			}
     		}

     		// create persistent node if not exists
     		Stat ipStat = null;
     		try {
     			ipStat = curator.checkExists().forPath(workerType + "/" + ip);
     		} catch (Exception ex) {
     			logger.error(ex.getMessage(), ex);
     		}
     		if (ipStat == null) {
     			try {
     				curator.create()
     						.withMode(CreateMode.EPHEMERAL)
     						.forPath(workerType + "/" + ip,
     								"0".getBytes(PropertiesCache.getInstance().get("charset.utf8")));
     			} catch (Exception ex) {
     				logger.error(ex.getMessage(), ex);
     			}
     		}
     		else{
     			try {
     				curator.setData().forPath(workerType + "/" + ip,
     						"0".getBytes(PropertiesCache.getInstance().get("charset.utf8")));
     			} catch (Exception ex) {
     				logger.error(ex.getMessage(), ex);
     			}
     		}
    }

    public void connect(String zkServer) {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        curator = CuratorFrameworkFactory.newClient(zkServer, retryPolicy);
        curator.start();
        try {
            curator.blockUntilConnected();
        } catch (InterruptedException ex) {
        	ex.printStackTrace();
            logger.error(ex.getMessage(), ex);
        }
    }

    public void disconnect(){
    	ThreadUtil.safeClose(executor);
		if (curator != null) {
			if(!StringUtil.isEmpty(workerType)){
				try {
					curator.setData().forPath(workerType + "/" + ip,
							"-1".getBytes(PropertiesCache.getInstance().get("charset.utf8")));
				} catch (Exception ex) {
					logger.error(ex.getMessage(), ex);
				}
			}
			CloseableUtils.closeQuietly(curator);
		}
    }

    public void increaseChannelCount(final String workerType){
        executor.execute(new Runnable(){
            @Override
            public void run() {
                String value = String.valueOf(channelCount.incrementAndGet());
                try{
                    curator.setData().forPath(workerType + "/" + ip, value.getBytes(PropertiesCache.getInstance().get("charset.utf8")));
                }catch(Exception ex){
                	ex.printStackTrace();
                    logger.error(ex.getMessage(), ex);
                    setIp(workerType, ip);
                }
            }
        });
    }

    public void decreaseChannelCount(final String workerType){
        executor.execute(new Runnable(){
            @Override
            public void run() {
                long count = channelCount.decrementAndGet();
                if(count < 0){
                    channelCount.set(0);
                }
                String value = String.valueOf(channelCount.get());
                try{
                    curator.setData().forPath(workerType + "/" + ip, value.getBytes(PropertiesCache.getInstance().get("charset.utf8")));
                    System.out.println("Compeleted to decrease to : " + value);
                }catch(Exception ex){
                	ex.printStackTrace();
                    logger.error(ex.getMessage(), ex);
                    setIp(workerType, ip);
                }
            }
        });
    }

}
